package com.yeming.flink.practice.state

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 第一种方法实现
 * 统计每个手机的呼叫时间间隔，时间单位是毫秒
 */
object TestKeyedState1 {

  def main(args: Array[String]): Unit = {
    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    //读取数据源
    val fileFile: String = getClass.getResource("/station.log").getPath
    val stream: DataStream[String] = streamEnv.readTextFile(fileFile)
    //数据计算，计算通话时间，开始时间和结束时间
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    streamLog.keyBy(_.callOut)
      .flatMap(new CallIntervalFunction).print("与上一次通话间隔：")

    //执行流计算
    streamEnv.execute()
  }

  class CallIntervalFunction extends RichFlatMapFunction[StationLog,(String,Long)]{

    //定义一个状态用于保存前一次呼叫的时间
    private var preCallTimeState:ValueState[Long] = _

    override def open(parameters: Configuration): Unit = {
      preCallTimeState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("pre", classOf[Long ]))
    }

    override def flatMap(value: StationLog, out: Collector[(String, Long)]): Unit = {
      var preCallTime = preCallTimeState.value()
      if (preCallTime == 0) {
        preCallTimeState.update(value.callTime)
//        out.collect((value.callOut,0))
      } else {
        //与上一次通话的时间间隔
        var interval = value.callTime - preCallTime
        // 输出数据
        out.collect((value.callOut, interval))
      }
    }
  }
}
